﻿using System;
using System.Collections.Generic;
using System.Threading;
using Needle.Bus.Engine.Utils;
using Needle.Bus.Messages;

namespace Needle.Bus.Engine
{
    public sealed class SinkProxy : EngineComponent, ISink
    {

        private readonly ISink _sink;

        private Predicate<IMessage> _filter;
        public Predicate<IMessage> Filter
        {
            get
            {
                ThrowIfDisposed();
                return _filter;
            }
            set
            {
                ThrowIfDisposed();
                var filter = _filter;
                while (Interlocked.CompareExchange(ref _filter, value, filter) != filter)
                    filter = _filter;
            }
        }

        #region constructors

        /// <summary>
        /// Initializes a new instance of the SinkFilter class.
        /// </summary>
        /// <param name="sink"></param>
        public SinkProxy(ISink sink) 
            : this(new NamedMutex(sink.Id + "Proxy"), sink, null) { }

        /// <summary>
        /// Initializes a new instance of the SinkFilter class.
        /// </summary>
        /// <param name="sink"></param>
        public SinkProxy(object mutex, ISink sink) 
            : this(mutex, sink, null) { }

        /// <summary>
        /// Initializes a new instance of the SinkFilter class.
        /// </summary>
        /// <param name="sink"></param>
        public SinkProxy(object mutex, ISink sink, Predicate<IMessage> filter)
            : base(mutex,sink.Id + "Proxy")
        {
            if (sink == null)
                throw new ArgumentNullException("sink", "sink is null.");

            _sink = sink;
            _filter = filter;
        }

        #endregion

        public void Enqueue(IMessage message)
        {
            if (!AcceptMessage(message))
                throw new MessageRejectedException();

            _sink.Enqueue(message);
        }

        public void Enqueue(IList<IMessage> messages)
        {
            if (!AcceptMessages(messages))
                throw new MessageRejectedException();

            _sink.Enqueue(messages);
        }

        public bool TryEnqueue(IMessage message)
        {
            if (AcceptMessage(message))
                return _sink.TryEnqueue(message);

            return false;
        }

        public bool TryEnqueue(IList<IMessage> messages)
        {
            if (AcceptMessages(messages))
                return _sink.TryEnqueue(messages);

            return false;
        }

        public IDisposable PrepareForEnqueue(IList<IMessage> messages)
        {
            if (!AcceptMessages(messages))
                throw new MessageRejectedException();

            return _sink.PrepareForEnqueue(messages);
        }

        public void Commit(object transaction)
        {
            ThrowIfDisposedOrNotOpen();
            _sink.Commit(transaction);
        }

        public void Abort(object transaction)
        {
            /* Don't throw if disposed, Abort always succeeds */
            _sink.Abort(transaction);
        }

        public int Count
        {
            get
            {

                return _sink.Count;
            }
        }

        private bool AcceptMessage(IMessage message)
        {
            if (IsDisposed)
                return false;

            var filter = _filter;
            return (filter == null || filter(message)) && !IsDisposed;
        }

        private bool AcceptMessages(IList<IMessage> messages)
        {
            if (IsDisposed)
                return false;

            var filter = _filter;
            if (filter == null)
                return IsDisposed;
            foreach (IMessage message in messages)
            {
                if (filter(message))
                    continue;
                else
                    return false;
            }
            return IsDisposed;
        }

        protected override void OnOpen(TimeSpan timeout)
        {
            // TODO: implement
        }

        protected override void OnAbort()
        {
            // TODO: implement
        }

        protected override void OnClose(TimeSpan timeout)
        {
            // TODO: implement
        }
    }

}
